/**
 * Copyright (C) 2017-2018 pinganfang, Inc. All Rights Reserved.
 */
package com.pingan.haofang.searchcloud.es.api.client;

import com.pingan.haofang.searchcloud.api.constants.EngineType;
import com.pingan.haofang.searchcloud.api.route.AliasUtils;
import com.pingan.haofang.searchcloud.api.route.IndexAliasRegister;
import com.pingan.haofang.searchcloud.api.route.IndexRoute;
import com.pingan.haofang.searchcloud.es.utils.EsUtils;
import org.apache.commons.collections.CollectionUtils;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.cluster.metadata.AliasMetaData;
import org.elasticsearch.common.collect.ImmutableOpenMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

/**
 * ES别名注册
 *
 * @author zhangbi617
 * @date 2017-07-07
 * @since 1.0.0
 */
@Component("esIndexAliasRegister")
public class EsIndexAliasRegister implements IndexAliasRegister {

    private static final Logger LOGGER = LoggerFactory.getLogger(EsIndexAliasRegister.class);

    @Autowired
    private EsClientFactory esClientFactory;

    /* (non-Javadoc)
     * @see com.pingan.haofang.searchcloud.api.route.IndexAliasRegister#register(com.pingan.haofang.searchcloud.api.route.IndexRoute)
     */
    @Override
    public boolean register(IndexRoute indexRoute) {
        if (indexRoute.getEngineType() != EngineType.ES.getValue()) {
            LOGGER.warn("engineType is not elasticsearch, return false");
            return false;
        }

        TransportClient client = esClientFactory.getByHost(indexRoute.getEngineHost());

        String aliasName = AliasUtils.convertAliasName(indexRoute.getCollectionRouteKey());

        ImmutableOpenMap<String, List<AliasMetaData>> oldAliasIndexMap = client.admin().indices()
                .prepareGetAliases(aliasName).execute().actionGet()
                .getAliases();

        List<String> oldIndicesNames = new ArrayList<>();
        Iterator<String> keysIt = oldAliasIndexMap.keysIt();
        while (keysIt.hasNext()) {
            oldIndicesNames.add(keysIt.next());
        }

        String realCollectionName = EsUtils.nameWithPrefix(indexRoute.getCollectionName());

        if (CollectionUtils.isEmpty(oldIndicesNames)) {
            client.admin().indices().
                    prepareAliases()
                    .addAlias(realCollectionName, aliasName)
                    .get();
        } else {
            client.admin().indices().
                    prepareAliases()
                    .removeAlias(oldIndicesNames.toArray(new String[oldIndicesNames.size()]), aliasName)
                    .addAlias(realCollectionName, aliasName)
                    .get();
        }


        LOGGER.info("alias created, collections:{}, alias:{}", realCollectionName,
                aliasName);

        return true;
    }

}
